#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include <stdint.h>
#include <amqp_tcp_socket.h>
#include <amqp.h>
#include <amqp_framing.h>

#include <assert.h>
#include <pthread.h>
#include "utils.h"

#include <pthread.h>

#include "cJSON.h"
#include <termios.h>
#include <sys/types.h>

#include <sys/stat.h>
#include <fcntl.h>
#include "amqp_config.h"
#include "uart.h"



static int connect_lost_flag = 0;
void *rabbitmq_send_thread(void)
{
	// nmea_msg gps;
    char rev_buf[1024] = "[{\"CO\":\"254\",\"FN\":\"\",\"FT\":1,\"HE\":17,\"HX\":\"1437248085\",\"LA\":39.665916442871094,\"LO\":119.059326171875,\"RE\":\"\",\"RT\":\"2020-10-21 09:46:35\",\"SP\":38.891998291015625,\"TE\":\"2020-10-21 09:46:35\"}]";
	// UploadMesTypeDef mes;
  int rc;
  int  status;
  
  amqp_socket_t *socket = NULL;
  amqp_connection_state_t conn;
  struct timeval tv;

  tv.tv_sec = 5;



   printf("[send]start thread\r\n");
   conn = amqp_new_connection();

  socket = amqp_tcp_socket_new(conn);
  if (!socket) {
    die("creating TCP socket");
  }

  status = amqp_socket_open_noblock(socket, s_hostname, s_port,&tv);
  if (status) {
	connect_lost_flag = 1;
    //die("opening TCP socket");
	printf("[send]Open TCP fail\n");
	return NULL;
  }

  die_on_amqp_error(amqp_login(conn, s_vhost, 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               s_user, s_pwd),
                    "Logging in");
  amqp_channel_open(conn, 1);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");

  
    amqp_basic_properties_t props;
    props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
    props.content_type = amqp_cstring_bytes("text/plain");
    props.delivery_mode = 2; /* persistent delivery mode */
   
  


	while(1)

	{

	//	r_count = uart_recv(uart_fd,rev_buf,sizeof(rev_buf));
	//	NMEA_GNRMC_Analysis(&gps,rev_buf);
	//	NMEA_GNVTG_Analysis(&gps,rev_buf);

		// mes.LO = gps.longitude/10000;
		// mes.LA = gps.latitude/10000;
		// mes.SP = gps.speed/1000;
		// mes.HE = gps.altitude;
		
		// str = json_package(JSON_OUT_UNFORMATE,&mes);

        rc = amqp_basic_publish(conn, 1, amqp_cstring_bytes(s_exchange),
                                    amqp_cstring_bytes(s_bindingkey),
                                     0, 
                                     0,
                                    &props, 
                                    amqp_cstring_bytes(rev_buf)
        );
		
        if(rc==AMQP_STATUS_HEARTBEAT_TIMEOUT || rc == AMQP_STATUS_SOCKET_ERROR)
         {
        
           connect_lost_flag = 1;
            
           break;
        } 
		if(connect_lost_flag == 1 ){
			break;
		
		}
        sleep(1);
		
	}
  
  if(connect_lost_flag == 1 ){
	  die_on_error(amqp_destroy_connection(conn), "Ending connection");
	  printf("[send] break thread!\r\n");
		return NULL;
	}
  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS),
                    "Closing channel");
  die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS),
                    "Closing connection");
  die_on_error(amqp_destroy_connection(conn), "Ending connection");
  printf("[send] closed thread!\r\n");
}

void *rabbitmq_reader_thread(void *arg)
{
  
  int  status;

  amqp_socket_t *socket = NULL;
  amqp_connection_state_t conn;

  amqp_bytes_t queuename;

	struct timeval tv;

  tv.tv_sec = 5;

  printf("[receive]start thread\r\n");
  conn = amqp_new_connection();

  socket = amqp_tcp_socket_new(conn);
  if (!socket) {
    die("creating TCP socket");
  }

  status = amqp_socket_open_noblock(socket, r_hostname, r_port,&tv);
  if (status) {
	  connect_lost_flag = 1;
    //die("opening TCP socket");
	printf("[receive]Open fail\n");
	return NULL;
  }

  die_on_amqp_error(amqp_login(conn, r_vhost, 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               r_user, r_pwd),
                    "Logging in");
  amqp_channel_open(conn, 1);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");

  {
    amqp_queue_declare_ok_t *r = amqp_queue_declare(
        conn, 1, amqp_empty_bytes, 0, 0, 0, 1, amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
    queuename = amqp_bytes_malloc_dup(r->queue);
    if (queuename.bytes == NULL) {
      fprintf(stderr, "Out of memory while copying queue name");
      return NULL;
    }
  }

  amqp_queue_bind(conn, 1, queuename, amqp_cstring_bytes(r_exchange),
                  amqp_cstring_bytes(r_bindingkey), amqp_empty_table);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Binding queue");

  amqp_basic_consume(conn, 1, queuename, amqp_empty_bytes, 0, 1, 0,
                     amqp_empty_table);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");

  {
    for (;;) {
      amqp_rpc_reply_t res;
      amqp_envelope_t envelope;

      amqp_maybe_release_buffers(conn);

      res = amqp_consume_message(conn, &envelope, &tv, 0);
      printf("amqp_consume_message : %d\n" ,res.reply_type);
      if (AMQP_RESPONSE_NORMAL != res.reply_type) 
      {
		connect_lost_flag = 1;
        break;
      }
      
      printf("reply_type %d ,Delivery %u, exchange %.*s routingkey %.*s\n",
             res.reply_type ,(unsigned)envelope.delivery_tag, (int)envelope.exchange.len,
             (char *)envelope.exchange.bytes, (int)envelope.routing_key.len,
             (char *)envelope.routing_key.bytes);

      if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
        printf("Content-type: %.*s\n",
               (int)envelope.message.properties.content_type.len,
               (char *)envelope.message.properties.content_type.bytes);
      }
      printf("----\n");

      amqp_dump(envelope.message.body.bytes, envelope.message.body.len);

      amqp_destroy_envelope(&envelope);
	  if(connect_lost_flag==1){
		  break;
	  }

    }
  }
 
  amqp_bytes_free(queuename);

  if(connect_lost_flag == 1 ){
	  die_on_error(amqp_destroy_connection(conn), "Ending connection");
	  printf("[receive]break thread\n");
	   return NULL;
	}


  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS),
                    "Closing channel");
  die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS),
                    "Closing connection");
  die_on_error(amqp_destroy_connection(conn), "Ending connection");
   printf("[receive] break thread\n");

  return 0;


}


int main(int argc,char *argv[])
{
	if(argc<2){
		printf("Usage: %s config_file\n",argv[0]);
		return 0;
	}
	if(0!=config_parse(argv[1])){
		return 0;
	}
	// uart init
	// const char* uart = "/dev/ttymxc2";
	//uart_fd = open(uart,O_RDWR|O_NOCTTY);
	// if(uart_fd==-1)
	// {
	// 	printf("open uart fail!\r\n");
	// }
	//set_opt(uart_fd,9600,8,'N',1);

        pthread_t rabbit_receive;
        while(1)
        {	
			connect_lost_flag = 0;	
            if(pthread_create(&rabbit_receive,NULL,rabbitmq_reader_thread,NULL))
            {
                perror("[ERROR] pthread create rabbit Fail.");
            }

            pthread_t rabbit_send;		
            if(pthread_create(&rabbit_send,NULL,rabbitmq_send_thread,NULL))
            {
                perror("[ERROR] pthread create rabbit Fail.");
            }
            
            if ( pthread_join ( rabbit_receive, NULL ) ) 
            {
                printf("error join thread.\n");
                
            }

            if ( pthread_join ( rabbit_send, NULL ) ) 
            {
                printf("error join thread.\n");
                
            }
            // pthread_cancel(rabbit_send);
            printf("[main]reconnecting! \n");
            sleep(1);
        }

         



}